package top.yehot.kettle.boot.core.task;

import java.io.File;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

import org.apache.commons.io.FileUtils;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.repository.AbstractRepository;
import org.pentaho.di.repository.RepositoryObjectType;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.InterruptableJob;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.UnableToInterruptJobException;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;

import lombok.extern.slf4j.Slf4j;
import top.yehot.cmm.util.spring.utils.SpringUtil;
import top.yehot.kettle.boot.core.SysConst;
import top.yehot.kettle.boot.core.config.KettleCfg;
import top.yehot.kettle.boot.core.execute.JobExecute;
import top.yehot.kettle.boot.core.execute.TransExecute;
import top.yehot.kettle.boot.core.utils.FileUtil;
import top.yehot.kettle.boot.core.utils.RepositoryUtil;
import top.yehot.kettle.boot.enums.GlobalParamEnum;
import top.yehot.kettle.boot.models.entity.GlobalParam;
import top.yehot.kettle.boot.models.entity.Task;
import top.yehot.kettle.boot.service.GlobalParamService;
import top.yehot.kettle.boot.service.TaskService;

/**
 * kettle脚本定时任务
 * 
 * @ClassName KettleJob
 * @author liuquan
 * @date 2022年5月13日 上午10:50:14
 */
@Slf4j
@DisallowConcurrentExecution
public class KettleJob implements InterruptableJob {

	private volatile Thread thisThread;
	
	@Override
	public void execute(JobExecutionContext context) throws JobExecutionException {
		// 获取当前线程
		thisThread = Thread.currentThread();

		// 获取TaskService
		TaskService taskService = SpringUtil.getBean(TaskService.class);
		// 获取JobDataMap
		JobDataMap jobData = context.getJobDetail().getJobDataMap();
		
		// 日志文件路径
		String logFilePath = jobData.getString(SysConst.TASK_LOG);
		// 任务ID
		int taskId = jobData.getInt(SysConst.TASK_ID);
		// 本次任务执行记录ID
		String recordId = jobData.getString(SysConst.TASK_RECORD_ID);
		
		String logText = "";
		try {
			// 获取任务信息
			Task task = taskService.getById(taskId);
			// 日志级别
			LogLevel logLevel = LogLevel.getLogLevelForCode(task.getLogLevel());
			// 获取参数
			Map<String, String> params = getParam(taskId + "", recordId);
			
			// 获取资源库
			AbstractRepository abstractRepository = RepositoryUtil.getRepository(task.getRepId());
			if (RepositoryObjectType.TRANSFORMATION.getTypeDescription().equals(task.getScriptType())) {
				logText = TransExecute.run(abstractRepository, task.getScriptPath(), task.getScriptName(), params, logLevel);
			} else if (RepositoryObjectType.JOB.getTypeDescription().equals(task.getScriptType())) {
				logText = JobExecute.run(abstractRepository, task.getScriptPath(), task.getScriptName(), params, logLevel);
			}
		} catch (Exception e) {
			String msg = "执行kettle脚本失败";
			log.error(msg, e);
			logText += msg + "\r\n" + e.getMessage();
			throw new JobExecutionException(e);
		} finally {
			writeFile(taskId, logFilePath, logText);
		}
	}

	@Override
	public void interrupt() throws UnableToInterruptJobException {
		thisThread.interrupt();
	}
	
	/**
	 * kettle 执行日志写入文件中
	 * 
	 * @param taskId 任务ID
	 * @param logFilePath 日志文件路径
	 * @param logText 日志内容
	 */
	private void writeFile(Integer taskId, String logFilePath, String logText) {
		try {
			// 判断日志文件夹是否存在，不存在创建文件夹
			File baseFile = new File(FileUtil.getParentPath(logFilePath));
			if (!baseFile.exists()) baseFile.mkdirs();
			
			// 日志写入文件
			FileUtils.writeStringToFile(new File(logFilePath), logText, KettleCfg.encoding.name(), false);
		} catch (Exception e) {
			log.error("任务ID:{}-执行kettle脚本，写入日志失败", taskId, e);
		}
	}
	
	/**
	 * 获取参数
	 */
	private Map<String, String> getParam(String taskId, String recordId) {
		Map<String, String> params = new HashMap<String, String>();
		// 全局参数类型
		String globalType = GlobalParamEnum.GLOBAL.getValue();
		
		// 获取TaskService
		GlobalParamService globalParamService = SpringUtil.getBean(GlobalParamService.class);
		// 获取[全局参数、任务私有参数]
		List<GlobalParam> glList = globalParamService.list(Wrappers.lambdaQuery(GlobalParam.class).in(GlobalParam::getCmmv, globalType, taskId));
		Optional.ofNullable(glList).ifPresent(gls -> {
			Map<String, List<GlobalParam>> tmap = glList.stream().collect(Collectors.groupingBy(GlobalParam::getCmmv));
			// 全局参数
			Optional.ofNullable(tmap.get(globalType)).ifPresent(mgls -> {
				mgls.stream().forEach(p -> params.put(p.getPkey(), p.getPvalue()));
			});
			// 任务私有参数
			Optional.ofNullable(tmap.get(taskId)).ifPresent(mgls -> {
				mgls.stream().forEach(p -> params.put(p.getPkey(), p.getPvalue()));
			});
		});
		params.put(SysConst.TASK_RECORD_ID, recordId);
		return params;
	}

}
